package com.visec.hdy.hadoop.storage;

import org.apache.log4j.Logger;

import com.visec.hdy.hadoop.standard.Analysis.Item;
import com.visec.hdy.hadoop.standard.Storage;
import com.visec.hdy.hadoop.standard.WriteContext;

/**
 * 按时间刷新数据存储代理实现类
 * 
 * @author hdy
 *
 * @param <K>
 * @param <V>
 */
public class TimeCacheStorageProxy<K, V> implements Storage<K, V> {
	private Storage<K, V> storage;
	private Thread t;
	final Logger l = Logger.getLogger(TimeCacheStorageProxy.class);

	public TimeCacheStorageProxy(Storage<K, V> storage) {
		this(storage, 1);
	}

	public TimeCacheStorageProxy(Storage<K, V> storage, int time) {
		this.storage = storage;
		t = new Cache(time);
		t.setDaemon(true);
	}

	@Override
	public void store(Item<K, V> item) {
		storage.store(item);
	}

	@Override
	public void open() {
		storage.open();
		t.start();
	}

	@Override
	public void setContext(WriteContext<K> context) {
		storage.setContext(context);
	}

	@Override
	public void flush() {
		storage.flush();
	}

	@Override
	public void close() {
		storage.close();
		if (t != null)
			t.interrupt();
	}

	public class Cache extends Thread {
		int time;

		public Cache(int time) {
			this.time = time;
		}

		@Override
		public void run() {
			while (!isInterrupted()) {
				try {
					synchronized (this) {
						wait(this.time * 1000L);
					}
					storage.flush();
				} catch (InterruptedException e) {
					interrupt();
				} catch (Exception e) {
					l.error("", e);
				}
			}
		}
	}
}
